

with tab_yl_oms_interceptorpiece_base as (
    select
         id
        ,list_id
        ,package_number
        ,data_collection_time
        ,scan_time
        ,scan_pda
        ,status
        ,waybill_no
        ,source
        ,type
        ,type_name
        ,registration_network_id
        ,registration_network_name
        ,registration_network_code
        ,registrant_id
        ,registrant_name
        ,registrant_code
        ,registration_time
        ,intercept_network_id
        ,intercept_network_name
        ,intercept_network_code
        ,intercept_id
        ,intercept_name
        ,intercept_code
        ,intercept_time
        ,intercept_num
        ,cancal_network_id
        ,cancal_network_name
        ,cancal_network_code
        ,cancal_id
        ,cancal_name
        ,cancal_code
        ,cancal_time
        ,registration_remark
        ,cancal_remark
        ,scan_type_code
        ,scan_type_name
        ,scan_network_id
        ,scan_network_name
        ,scan_network_code
        ,update_time
        ,row_number() over(partition by id order by update_time desc) as row_id
        ,date_format(registration_time,'yyyy-MM-dd') as dt
    from jms_ods.yl_oms_interceptorpiece_di
    where dt >= date_add('{{ execution_date | cst_ds }}', -90)
    and dt <= date_add('{{ execution_date | cst_ds }}', 0)
)

insert overwrite table jms_dwd.dwd_yl_oms_interceptorpiece_base_dt partition(dt)
select
     id
    ,list_id
    ,package_number
    ,data_collection_time
    ,scan_time
    ,scan_pda
    ,status
    ,waybill_no
    ,source
    ,type
    ,type_name
    ,registration_network_id
    ,registration_network_name
    ,registration_network_code
    ,registrant_id
    ,registrant_name
    ,registrant_code
    ,registration_time
    ,intercept_network_id
    ,intercept_network_name
    ,intercept_network_code
    ,intercept_id
    ,intercept_name
    ,intercept_code
    ,intercept_time
    ,intercept_num
    ,cancal_network_id
    ,cancal_network_name
    ,cancal_network_code
    ,cancal_id
    ,cancal_name
    ,cancal_code
    ,cancal_time
    ,registration_remark
    ,cancal_remark
    ,scan_type_code
    ,scan_type_name
    ,scan_network_id
    ,scan_network_name
    ,scan_network_code
    ,update_time
    ,dt
from tab_yl_oms_interceptorpiece_base
where dt >= date_add('{{ execution_date | cst_ds }}', -90)
and dt <= date_add('{{ execution_date | cst_ds }}', 0)
and row_id = 1
distribute by dt;